package com.simon.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.IntStream;

public class FlowMain {

    public static void main(String[] args) throws InterruptedException {

        // sample();

        processor();
    }

    private static void processor() throws InterruptedException {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        Flow.Subscriber<String> oneSubscriber = new OneSubscriber();
        // 中间流
        CustomProcessor oneProcessor = new CustomProcessor("oneProcessor");
        CustomProcessor twoProcessor = new CustomProcessor("twoProcessor");
        publisher.subscribe(oneProcessor);
        oneProcessor.subscribe(twoProcessor);
        twoProcessor.subscribe(oneSubscriber);

        IntStream.rangeClosed(1, 10).mapToObj(String::valueOf).forEach(publisher::submit);

        publisher.close();

        Thread.sleep(20000);
    }

    private static void sample() throws InterruptedException {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        Flow.Subscriber<String> oneSubscriber = new OneSubscriber();
        Flow.Subscriber<String> twoSubscriber = new TwoSubscriber();

        publisher.subscribe(oneSubscriber);
        publisher.subscribe(twoSubscriber);

        IntStream.rangeClosed(1, 10).mapToObj(String::valueOf).forEach(publisher::submit);

        publisher.close();

        Thread.sleep(20000);
    }

}
